/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.thread;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.io.FileUtils;

import cn.com.yang.utils.ListUtils;

/**
 * @Model cn.com.yang.thread.AbstractBigFileTools
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月21日-上午10:54:53
 */
/**
 * @desc 处理大文件的工具类
 * @author cmc
 * @email chenmingchuan@all-in-data.com
 * @date 2019年11月25日 下午2:30:21
 */
public abstract class AbstractBigFileTools {
	
	private static final String ENCODING = "utf-8" ;
	//处理文件段大小，默认20000
	private int pageSize = 20000 ; 
	//读取当前CPU个数，决定线程池的大小
	private final static int POOLSIZE = Runtime.getRuntime().availableProcessors() ;
	//线程池
	private ExecutorService executor = Executors.newFixedThreadPool(POOLSIZE) ;
	//任务队列
	private ArrayBlockingQueue<List<String>> taskQueue = new ArrayBlockingQueue<>(POOLSIZE) ;
	
	private AtomicBoolean isFinsh = new AtomicBoolean(false) ;
	
	public AbstractBigFileTools(int pageSize) {
		super();
		this.pageSize = pageSize;
	}

	public AbstractBigFileTools() {
		super();
	}

	/**
	 * <p>功能: 把文件拆分成size大小分别处理，单线程 </p>
	 * @param file
	 * @param size
	 * @throws IOException 
	 */
	public void  handle(File file) throws IOException {
		int index = 1 ;
		long start = System.currentTimeMillis() ;
//		List<String> lines = FileCopyTools.copyToLines(file, ENCODING, true, index, pageSize) ;
		List<String> lines = FileUtils.readLines(file,ENCODING);
		while(!ListUtils.isEmpty(lines)) {
			handleInternal(lines, index) ;
			System.out.println("正在处理" + index + "-" + (index+pageSize-1) + "处数据，文件读取耗时：" 
					+ (System.currentTimeMillis()-start));
			index = index + pageSize ;
			lines.clear();
			start = System.currentTimeMillis() ;
//			lines = FileCopyTools.copyToLines(file, ENCODING, true, index, pageSize) ;
			lines = FileUtils.readLines(file,ENCODING);
		}
	}
	
	/**
	 * <p>功能: 把文件拆分成size大小分别处理，多线程 </p>
	 * @param file
	 * @param size
	 * @throws IOException 
	 */
	public void  handleSync(File file) throws IOException {
		
//		//打印日志
//		executor.execute(() ->{
//			printLogs() ;
//		});
		
		//将大文件拆分后放到任务队列
		executor.execute(() ->{
			try {
				putTasks(file) ;
			} catch (IOException | InterruptedException e) {
				e.printStackTrace();
			}
		}) ;
		//处理业务
		handleSyncInternal() ;
		
		executor.shutdown();
		
		while(!executor.isTerminated()) {
		}
		
		System.out.println("==任务执行完毕==");
	}


//	/**
//	 * <p>功能: 打印日志</p>
//	 */
//	private void printLogs() {
//		while(!isTerminated()) {
//			try {
//				TimeUnit.SECONDS.sleep(3);
//				System.out.println("任务执行状态：" + executor);
//			} catch (InterruptedException e) {
//				e.printStackTrace();
//			}
//		}
//	}

	/**
	 * <p>功能: 多线程处理具体任务</p>
	 */
	private void handleSyncInternal() {
		try {
			AtomicInteger startIndex = new AtomicInteger(1) ;
			while(!isFinsh.get()) {
				List<String> lines = taskQueue.poll(3, TimeUnit.SECONDS) ;
				executor.execute(() ->{
					int index = startIndex.getAndAdd(pageSize) ;
					System.out.println("正在处理" + index + "-" + (index+pageSize-1) + "处数据..." );
					handleInternal(lines, index) ;
				});
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	/**
	 * <p>功能: 往任务队列里放入任务</p>
	 * @param file
	 * @throws IOException 
	 * @throws InterruptedException 
	 */
	private void putTasks(File file) throws IOException, InterruptedException {
		List<String> lines = new ArrayList<String>();
		BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), ENCODING));
		try
		{
			String line = null;
			while ((line = reader.readLine()) != null)
			{
				lines.add(line) ;
				//如果list大于pageSize放入任务队列
				if(lines.size() >= pageSize) {
					taskQueue.put(lines);
					lines = new ArrayList<String>();
				}
			}
			//放入最后剩余部分
			taskQueue.put(lines);
		}
		finally
		{
			System.out.println("==生产线程结束了==");
			isFinsh.set(true);
			if (reader != null) reader.close();
		}
	}

	/**
	 * <p>功能: 具体业务，有子类实现</p>
	 * @param lines
	 */
	protected abstract void handleInternal(List<String> lines, int index) ;
	
	/**
	 * <p>功能: 批量写入文件</p>
	 * @param lines
	 * @param file
	 * @param size 每次写入的行数
	 */
	protected void batchWrite(List<String> lines, File file, int size) {
		int lineSize = 0 ;
		StringBuilder sb = new StringBuilder() ;
		for(String line : lines) {
			if(lineSize >= size) {
				write(file, sb.toString()) ;
				sb = new StringBuilder() ;
				lineSize = 0 ;
			}
			sb.append(line + "\n") ;
			lineSize ++ ;
		}
		write(file, sb.toString()) ;
	}
	
	/**
	 * <p>功能: 写入文件</p>
	 * @param string
	 */
	protected void write(File file, String str) {
		try {
			FileUtils.writeStringToFile(file, str , "utf-8", true);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
}

